package cn.ycc1.boot3kafka.service;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    /**
     * 监听 demo-topic
     */
    @KafkaListener(topics = "${app.kafka.topic.demo}", groupId = "${spring.kafka.consumer.group-id}")
    public void consumeMessage(String message) {
        System.out.println("收到消息: " + message);
        // 这里可以添加业务处理逻辑
    }

    /**
     * 监听 user-topic
     */
    @KafkaListener(topics = "${app.kafka.topic.user}", groupId = "${spring.kafka.consumer.group-id}")
    public void consumeUserMessage(String message) {
        System.out.println("收到用户消息: " + message);
        // 解析 JSON 并处理用户消息
        try {
            // 这里可以添加 Jackson 解析逻辑
            System.out.println("处理用户消息: " + message);
        } catch (Exception e) {
            System.err.println("解析用户消息失败: " + e.getMessage());
        }
    }

    /**
     * 或者使用 ConsumerRecord 获取完整信息
     */
    @KafkaListener(topics = "${app.kafka.topic.demo}", groupId = "${spring.kafka.consumer.group-id}")
    public void consumeWithRecord(ConsumerRecord<String, String> record) {
        System.out.println("Received message: " + record.value());
        System.out.println("Key: " + record.key());
        System.out.println("Partition: " + record.partition());
        System.out.println("Offset: " + record.offset());
        System.out.println("Headers: " + record.headers());
    }
}
